package com.c6906.vdam.dps.runner;

import com.alibaba.fastjson.JSON;
import com.c6906.vdam.dps.config.Rocket;
import com.c6906.vdam.dps.config.redis.RedisUtils;
import com.c6906.vdam.dps.entity.TerminalBusinessInfo;
import com.c6906.vdam.dps.entity.TerminalData;
import com.c6906.vdam.dps.service.TerminalBusinessInfoService;
import com.c6906.vdam.dps.service.TerminalDataService;
import com.c6906.vdam.dps.service.TerminalHistoryDetailsService;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.codec.binary.Base64;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.SimpleConsumer;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import javax.annotation.Resource;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @Author : TAOHUIQI
 * @Date: 2024/6/20 21:02
 * @Description:多线程消费者-SimpleConsumer模式
 */
@Order(1)
@Component
public class VehicleConsumerRunner implements CommandLineRunner {
    private static final Logger logger = LoggerFactory.getLogger(VehicleConsumerRunner.class);
    private static final String PRE_REDIS_KEY = "SN";
    @Resource
    RedisUtils redisUtils;
    @Resource
    Rocket rocket;
    @Resource
    TerminalBusinessInfoService terminalBusinessInfoService;
    @Resource
    TerminalHistoryDetailsService terminalHistoryDetailsService;
    @Resource
    TerminalDataService terminalDataService;

    @Override
    public void run(final String... args) {
        logger.info("轨迹数据消费线程池启动......");
        int poolSize = rocket.getPoolSize();
        ExecutorService executor = Executors.newFixedThreadPool(poolSize);
        for (int i = 0; i < poolSize; i++) {
            logger.info("轨迹数据消费线程池:{}", i);
            executor.submit(() -> {
                do {
                    SimpleConsumer consumer = null;
                    try {
                        final ClientServiceProvider provider = ClientServiceProvider.loadService();
                        String endpoints = rocket.getEndpoints();
                        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                                .setEndpoints(endpoints)
                                .build();
                        // RocketMQ的tag
                        String tag = "*";
                        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
                        // RocketMQ的ConsumerGroup
                        String consumerGroup = rocket.getVehicleConsumerGroup();
                        //设置循环消息频次
                        Duration awaitDuration = Duration.ofSeconds(30);
                        // RocketMQ的Topic
                        String topic = rocket.getVehicleTopic();
                        consumer = provider.newSimpleConsumerBuilder()
                                .setClientConfiguration(clientConfiguration)
                                .setConsumerGroup(consumerGroup)
                                .setAwaitDuration(awaitDuration)
                                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                                .build();
                        // 每次获取消息条数
                        int maxMessageNum = rocket.getBatchSize();
                        // 设置消息接收后的不可见持续时间
                        Duration invisibleDuration = Duration.ofSeconds(1000);
                        do {
                            final List<MessageView> messages = consumer.receive(maxMessageNum, invisibleDuration);
                            logger.info("获取消息{}条", messages.size());
                            if (!messages.isEmpty()) {
                                // 定义一个List，用户存放单次消费的记录，用于批量入库
                                List<TerminalData> dataList = new ArrayList<>();
                                for (MessageView message : messages) {
                                    ObjectMapper objectMapper = new ObjectMapper();
                                    try {
                                        String base64String = objectMapper.writeValueAsString(message.getBody());
                                        //消息Body转byte数组
                                        byte[] decodeBytes = Base64.decodeBase64(base64String);
                                        String msgString = new String(decodeBytes);
                                        TerminalData terminalData = JSON.parseObject(msgString, TerminalData.class);
                                        //根据设备编号判断redis是否存在
                                        TerminalBusinessInfo terminalBusinessInfo = redisUtils.terminalInfoIsExistFormRedis(PRE_REDIS_KEY + terminalData.getTerminalSn());
                                        if (null != terminalBusinessInfo && !ObjectUtils.isEmpty(terminalBusinessInfo.getTerminalSn())) {
                                            //填充DM数据设备基本信息
                                            generateTerminalData(terminalData, terminalBusinessInfo, terminalData.getTerminalSn());
                                            if (terminalData.getInsertFlag() == 0) {
                                                dataList.add(terminalData);
                                                logger.info("合法数据:{}", JSON.toJSONString(terminalData));
                                            }
                                        } else {
                                            logger.debug("非法数据:{}", JSON.toJSONString(terminalData));
                                        }
                                        consumer.ack(message);
                                    } catch (JsonProcessingException e) {
                                        logger.error("MQ消费反序列化失败, messageId={}", message.getMessageId());
                                    } catch (ClientException e) {
                                        logger.error("Failed to ack message, messageId={}", message.getMessageId());
                                    }
                                }
                                //批量入库到DM
                                terminalDataService.batchInsertDataMQToDM(dataList);
                                //保存两条最新的数据到redis
                                redisUtils.saveLastDatasForHistory(2, dataList);
                                terminalBusinessInfoService.saveTerminalDataToRedis(dataList);
                                //性能问题 终端告警处理程序速度优化
//                                terminalHistoryDetailsService.handleTerminalHistoryDetails(dataList);
                            }
                        } while (true);
                    } catch (Exception e) {
                        logger.error("轨迹数据消费异常", e);
                    } finally {
                        if (null != consumer) {
                            consumer.close();
                            logger.info("轨迹数据Consumer关闭");
                        } else {
                            logger.info("未创建Consumer，无需关闭");
                        }
                    }
                } while (true);
            });
        }
        logger.info("轨迹数据消费线程池启动结束......");
    }

    private void generateTerminalData(TerminalData terminalData, TerminalBusinessInfo terminalBusinessInfo, String terminalSn) {
        if (ObjectUtils.isEmpty(terminalBusinessInfo.getVehPlateNum())) {
            logger.error("设备{}的车牌为空，数据无法入库...", terminalSn);
            terminalData.setInsertFlag(1);
        } else if (ObjectUtils.isEmpty(terminalBusinessInfo.getVehiclesCode())) {
            logger.error("设备{}的车辆编码为空，数据无法入库...", terminalSn);
            terminalData.setInsertFlag(1);
        } else {
            terminalData.setRegionCode(terminalBusinessInfo.getRegionCode());
            terminalData.setRegionName(terminalBusinessInfo.getRegionName());
            terminalData.setVehiclesCode(terminalBusinessInfo.getVehiclesCode());
            terminalData.setVehPlateNum(terminalBusinessInfo.getVehPlateNum());
            terminalData.setOwnUnitId(terminalBusinessInfo.getOwnUnitId());
            terminalData.setOwnUnitName(terminalBusinessInfo.getOwnUnitName());
            terminalData.setUseUnitId(terminalBusinessInfo.getUseUnitId());
            terminalData.setUseUnitName(terminalBusinessInfo.getUseUnitName());
            terminalData.setVehUseType(terminalBusinessInfo.getVehUseType());
            terminalData.setVehOwnType(terminalBusinessInfo.getVehOwnType());
            terminalData.setVehPropId(terminalBusinessInfo.getVehPropId());
            terminalData.setPlateColorName(terminalBusinessInfo.getPlateColorName());
            terminalData.setInsertFlag(0);
        }
    }
}
